package com.iussoft.receiveData.send.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.apache.exception.BusinessException;
import com.apache.tools.DateUtils;
import com.apache.tools.NumberUtils;
import com.apache.tools.StrUtil;
import com.iussoft.analyze.utils.DataAnalyzeUtil;
import com.iussoft.model.DataParams;
import com.iussoft.receiveData.send.ReceiveDataManager;
import com.iussoft.tools.AesUtil;
import com.iussoft.tools.HttpService;
import com.iussoft.tools.LoggerUtil;
import com.iussoft.tools.XMLUtil;
import org.apache.commons.collections.map.CaseInsensitiveMap;
import org.springframework.stereotype.Service;
import org.springframework.util.DigestUtils;

import java.net.URLEncoder;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @description: 接收服务业务接口实现
 * @author: Hou Dayu
 * @date: Created in 2020/3/16
 */
@Service("receiveDataManager")
public class ReceiveDataManagerImpl extends SuperSearchManager implements ReceiveDataManager {

    /**
     * 解析接收到的信息实现方法
     */
    public void receiveDataMsg(DataParams params) {
        String rowkey = params.getTraceId();
        String sourceMsg = params.getParamsToStr("sourceMsg");
        String interfaceId = StrUtil.doNull(params.getParamsToStr("infterfaceId"), "");
        if("post".equals(params.getParamsToStr("requestType"))){
            JSONObject faceInfo = getInterfaceInfo(interfaceId);
            params.setParams("sysEname",faceInfo.getString("sysEname"));
            if("11".equalsIgnoreCase(faceInfo.getString("exchangeMode"))) {
                params.setParams("dataType", "req");
            } else {
                params.setParams("dataType", "resp");
            }
        }
        Map<String,Object> sourceMap = null;
        String sourceFile = params.getParamsToStr("sourceFile");
        Map<String,String> rules = checkMessage(sourceMsg, interfaceId, sourceFile);
        JSONObject source = sendMsgTool.getRequestMessage(sourceMsg);
        try {
            if(source == null){
                return;
            }
            if(sourceMap ==null || sourceMap.isEmpty()){
                sourceMap = new HashMap<>();
                sourceMap.putAll(source);
            }
            JSONObject result = dataAnalyze(rules,sourceMap,interfaceId,params);

            if(!"1".equals(params.getParamsToStr("ifTest"))) {
                dataSave(params, result, params.getParamsToStr("sysEname"));
            }else{
                LoggerUtil.info("挡板数据，不保存解析结果！");
            }
        } catch (Exception e){
            LoggerUtil.error("解析报文出错",e);
            if(e instanceof BusinessException){
                throw new BusinessException(e.getMessage());
            } else {
                LoggerUtil.warnSource("exception = {}; exeSys = send; infoType = exception; method = receiveDataMsg; traceId = {}; requestTime = {}; requestIp = {}; rowkey = {}; sourceMessage = {}; errorMessage = {}; level = {}", e.getClass().getSimpleName(), params.getParamsToStr("traceId"), DateUtils.Now.fmt_yyyyMMdd_HHmmss(), params.getParamsToStr("requestUseIp"), rowkey, sourceMsg.replaceAll("\n", "").replaceAll("\r", "").replaceAll(";", "；"), e.getMessage(), "ERROR");
            }
        }
    }

    /**
     * 被动推送的数据接入方式实现方法
     */
    @Override
    public void receiveDatapPassive(DataParams params) {

    }

    /**
     * 验证报文规则
     */
    private Map<String,String> checkMessage(String sourceMsg, String interfaceId, String sourceFile){
        LoggerUtil.info("数据预处理解阶段开始...");
        Map<String,String> rules = getRuleSite(interfaceId);
        if(!rules.isEmpty()){//验证解析规则
            if (StrUtil.isNotNull(rules.get("fileSuffix")) && StrUtil.isNotNull(sourceFile)) {//验证文后缀名
                if (!sourceFile.endsWith(rules.get("fileSuffix"))) {
                    throw new BusinessException("文件的后缀名不符合规则");
                }
            }
            if(StrUtil.isNotNull(rules.get("msgFormat"))){//验证报文格式
                String mf = rules.get("msgFormat");
                if("xml".equalsIgnoreCase(mf) && !XMLUtil.isXmlDocument(sourceMsg)){
                    throw new BusinessException("报文格式不符合xml规则");
                } else if("json".equalsIgnoreCase(mf) && !sendMsgTool.isJson(sourceMsg)){
                    throw new BusinessException("报文格式不符合json规则");
                }
            }
//            System.out.println("________----------------------->>>>"+sourceMsg.getBytes().length);
            double size = NumberUtils.divide(sourceMsg.getBytes().length, 1024, 4);
//            int size = sourceMsg.getBytes().length/1024;//变成Kb
            if(StrUtil.isNotNull(rules.get("msgSize"))){//验证报文大小
                String msgSize = rules.get("msgSize");
                double msize = Double.parseDouble(msgSize);
                if(size > msize){
                    throw new BusinessException("报文大小超过设定值("+msgSize+")kb");
                }
            } else {
                if(size > 1024){
                    throw new BusinessException("报文大小超过1M,需要人工处理");
                }
            }
        }
        return rules;
    }

    private JSONObject dataAnalyze(Map<String,String> rules, Map<String,Object> sourceMap, String interfaceId, DataParams params){
        LoggerUtil.info("数据解析阶段开始...");
        String dataType = "'"+params.getParamsToStr("dataType")+"'";
        Map<String,Object> formats = (Map)getInterfaceParams(interfaceId,dataType).get("body");
        String matchRule = "";
        String exceptionRule = "";
        if(rules != null && StrUtil.isNotNull(rules.get("matchRule"))){//验证匹配规则
            matchRule = rules.get("matchRule");
            exceptionRule = rules.get("exceptionRule");
        }

        String traceId = params.getParamsToStr("traceId");
        if(null == formats || formats.isEmpty()){
            LoggerUtil.warnSource("traceId = {}; exeSys = send; infoType = warning; errorMessage = 没有配置解析报文",traceId);
            /*JSONObject result = new JSONObject();
            result.putAll(sourceMap);*/
            return new JSONObject();
        }
        String rootName = String.valueOf(sourceMap.get("rootName"));
        if(!StrUtil.isNull(rootName)){
            //说明是xml
            LoggerUtil.info("开始解析XML格式，根节点="+rootName);
            Map<String,Object> tempMap = (Map)formats.get(rootName);
            formats.clear();
            formats.putAll((Map)tempMap.get("children"));
        }
        Map<String,Object> data = new HashMap<>();
        if("02".equals(matchRule)){//是否忽略大小写
            formats = new CaseInsensitiveMap(formats);
        }
        JSONObject result = DataAnalyzeUtil.dataAnalyze(sourceMap,formats,traceId,matchRule,exceptionRule,true);
        return result;
    }

    private void dataSave(DataParams params, JSONObject data,String infoType){
        LoggerUtil.infoSource("数据存处阶段开始...; 解析结果:{}",data.toJSONString());
        if(data.isEmpty() || data.size()==1 && data.containsKey("rowkeyStr")){
            return ;
        }
        String interfaceCode = StrUtil.doNull(params.getParamsToStr("interfaceCode"),"");
        Map<String,Object> esinfo = new HashMap();
        esinfo.put("sourceCode", StrUtil.doNull(params.getParamsToStr("requestCode"),""));
        esinfo.put("interfaceCode", interfaceCode);
        if(StrUtil.isNotNull(infoType)){
            esinfo.put("infoType", infoType);
        }
        esinfo.put("traceId", params.getParamsToStr("traceId"));
        esinfo.put("createdTime", DateUtils.strToLong(DateUtils.Now.fmt_yyyyMMdd_HHmmss(), DateUtils.FmtStr.yyyyMMdd_HHmmss));
        boolean mark = false;
        long rkTime = System.currentTimeMillis();
        JSONObject defVal = null;
        if(StrUtil.isNotNull(params.getParamsToStr("defaultColumn"))){//默认列
            defVal = JSON.parseObject(params.getParamsToStr("defaultColumn"));
        }
        String index_name = analyzeIndexName + "-" + DateUtils.Now.fmt_yyyy();
        boolean ifKafka = kafkaSimpleSender.getKafkaEnabled();
        String reqHbaseUrl = null;
        if(ifKafka) {
            reqHbaseUrl = esManager.getRequestUrl("search-server");
            String hbaseTabName = StrUtil.doNull(StrUtil.isNull(interfaceCode) ? "" : "hb_" + interfaceCode, getHbaseTableName(2));
            esinfo.put("tableName", hbaseTabName);
        }else {
            esinfo.put("tableName", index_name);
        }
        List<Map<String,Object>> dataList = new ArrayList();
        if(null != data && !data.isEmpty()) {
            if(StrUtil.isNotNull(data.getString("rowkeyStr"))) {
                String rowkeyStr = interfaceCode + data.getString("rowkeyStr")+rkTime;
                String rowkey = rollbackStr(DigestUtils.md5DigestAsHex(rowkeyStr.getBytes()));
                esinfo.put("rowkey", rowkey);
                Map ddm = new HashMap();
                for (String key : data.keySet()) {
                    if ("rowkeyStr".equalsIgnoreCase(key)) {
                        continue;
                    }
                    Object obj = data.get(key);
                    DataAnalyzeUtil.getMapToHbases(params.getParamsToStr("mergeParams"),ddm,obj);
                    /*Map ddm2 = (Map)getMapToHbase(params.getParamsToStr("mergeParams"),ddm,obj);
                    if(ddm2 !=null){
                        ddm.putAll(ddm2);
                    }*/
                }
                if (!ddm.isEmpty()) {
                    if(null != defVal){
                        ddm.putAll((Map)defVal);
                    }
                    if(ifKafka) {
                        String url = reqHbaseUrl + "hbase/write/" + rowkey + "/" + esinfo.get("tableName") + "/";
                        String msg = AesUtil.getInstance().encryptEcbMode(JSON.toJSONString(ddm));
                        HttpService.sendHttp(url, "post", null, msg, "json");
                    } else {
                        esinfo.put("message", JSON.toJSONString(ddm));
                    }
                    mark = esManager.saveInfoForEs(index_name, analyzeIndexName, rowkey, esinfo);
                }

                LoggerUtil.warnSource("traceId = {}; exeSys = send; processingStage = {}; rowkey = {}; interfaceCode = {}; requestIp = {}; requestCode = {}; processingResult = {}; infoType = info;",params.getParamsToStr("traceId"), "storage",rowkey, interfaceCode, params.getParamsToStr("requestUseIp"), params.getParamsToStr("requestCode"), String.valueOf(mark));
                if(mark){
                    sendDataSync(params,esinfo,dataList);
                }
            } else {
                List<Map> list = null;
                Map ddm = new HashMap();
                for (String key : data.keySet()) {
                    if ("rowkeyStr".equalsIgnoreCase(key)) {
                        continue;
                    }
                    Object obj = data.get(key);
                    DataAnalyzeUtil.getMapToHbases(params.getParamsToStr("mergeParams"),ddm,obj);
                    if(ddm.containsKey("dataList")){
                        list = (List)ddm.get("dataList");
                        ddm.remove("dataList");
                    }
                   // list = (List)getMapToHbase(params.getParamsToStr("mergeParams"),null,obj);
                }
                if(null != list && !list.isEmpty()){
                    for(int i=0;i<list.size();i++){
                        Map map = list.get(i);
                        Map<String,Object> resultData = new HashMap();
                        resultData.putAll(esinfo);
                        String rowkeyStr = interfaceCode + map.get("rowkeyStr")+rkTime;
                        String rowkey = rollbackStr(DigestUtils.md5DigestAsHex(rowkeyStr.getBytes()));
                        resultData.put("rowkey", rowkey);
                        map.remove("rowkeyStr");
                        Map dmap = new HashMap();
//                        if(map.size() > 1 ){
//                            dmap = (Map)getMapToHbase(dmap,map);
//                        } else {
                        DataAnalyzeUtil.getMapToHbase(dmap,"", map);
//                        }
                        if(null != defVal){
                            dmap.putAll((Map)defVal);
                        }
                        if(kafkaSimpleSender.getKafkaEnabled()) {
                            String url = reqHbaseUrl + "hbase/write/" + rowkey + "/" + esinfo.get("tableName") + "/";
                            String msg = AesUtil.getInstance().encryptEcbMode(JSON.toJSONString(dmap));
                            HttpService.sendHttp(url, "post", null, msg, "json");
                            dataList.add(resultData);
                        } else {
                            resultData.put("message",JSON.toJSONString(dmap));
                            dataList.add(resultData);
                        }
                    }
                    mark = esManager.batchSaveInfos(index_name, analyzeIndexName,"doc",dataList);
                    LoggerUtil.warnSource("traceId = {}; exeSys = send; processingStage = {}; rowkey = {}; interfaceCode = {}; requestIp = {}; requestCode = {}; processingResult = {}; infoType = info;",params.getParamsToStr("traceId"), "storage",params.getParamsToStr("traceId"), interfaceCode, params.getParamsToStr("requestUseIp"), params.getParamsToStr("requestCode"), String.valueOf(mark));
                    if(mark){
                        sendDataSync(params,null, dataList);
                    }
                } else if(!ddm.isEmpty()){
                    if(null != defVal){
                        ddm.putAll((Map)defVal);
                    }
                    String rowkeyStr = interfaceCode + rkTime;
                    String rowkey = rollbackStr(DigestUtils.md5DigestAsHex(rowkeyStr.getBytes()));
                    esinfo.put("rowkey", rowkey);
                    if(ifKafka) {
                        String url = reqHbaseUrl + "hbase/write/" + rowkey + "/" + esinfo.get("tableName") + "/";
                        String msg = AesUtil.getInstance().encryptEcbMode(JSON.toJSONString(ddm));
                        HttpService.sendHttp(url, "post", null, msg, "json");
                    } else {
                        esinfo.put("message", JSON.toJSONString(ddm));
                    }
                    mark = esManager.saveInfoForEs(index_name, analyzeIndexName, rowkey, esinfo);
                    LoggerUtil.warnSource("traceId = {}; exeSys = send; processingStage = {}; rowkey = {}; interfaceCode = {}; requestIp = {}; requestCode = {}; processingResult = {}; infoType = info;",params.getParamsToStr("traceId"), "storage",rowkey, interfaceCode, params.getParamsToStr("requestUseIp"), params.getParamsToStr("requestCode"), String.valueOf(mark));
                    if(mark){
                        sendDataSync(params,esinfo,dataList);
                    }
                }
            }
        }
        if(!mark){
            throw new BusinessException("数据存储阶段失败");
        }
    }

    /**
     * 数据同步操作
     */
    private void sendDataSync(DataParams params, Map<String,Object> esinfo, List<Map<String,Object>> dataList){
        if("1".equals(StrUtil.doNull(getConfig("data.sync.enabled"),"0"))) {
            String interfaceCode = StrUtil.doNull(params.getParamsToStr("interfaceCode"), "");
            if (kafkaSimpleSender.getKafkaEnabled()) {//开启了Kafka同步
                if(StrUtil.isEmpty(dataList)){
                    kafkaSimpleSender.sendMsg(sendMsgTopic(99), "0", JSON.toJSONString(esinfo));
                } else {
                    dataList.forEach(item->kafkaSimpleSender.sendMsg(sendMsgTopic(99), "0", JSON.toJSONString(item)));
                }
            } else {
                String reqUrl = esManager.getRequestUrl("");
                if (StrUtil.isNotNull(reqUrl)) {
                    Map<String, String> headParams = new HashMap();
                    if (StrUtil.isEmpty(dataList)) {//单条同步
                        headParams.put("datamessage", AesUtil.getInstance().encryptEcbMode(JSON.toJSONString(esinfo)));
                        HttpService.sendPostHttp(reqUrl + "/sendSync/" + esinfo.get("tableName"), null, headParams);
                    } else {//批量同步
                        Map<String, Object> data = dataList.get(0);
                        String msg = AesUtil.getInstance().encryptEcbMode(JSON.toJSONString(dataList));
                        headParams.put("datamessage", msg);
                        HttpService.sendPostHttp(reqUrl + "/sendSync/" + data.get("tableName") + "?type=batch", null, headParams);
                    }
                    LoggerUtil.infoSource("traceId = {}; exeSys = send; processingStage = {}; interfaceCode = {}; requestCode = {}; processingResult = {}; infoType = info;", params.getParamsToStr("traceId"), "sendMsg", interfaceCode, params.getParamsToStr("requestCode"), "消息发送成功");
                } else {
                    LoggerUtil.warnSource("exception = {}; exeSys = send; infoType = exception; method = receiveDataMsg; traceId = {}; requestTime = {}; requestIp = {}; sourceMessage = {}; errorMessage = {}; level = {}", "sendDataSync", params.getParamsToStr("traceId"), DateUtils.Now.fmt_yyyyMMdd_HHmmss(), params.getParamsToStr("requestUseIp"), params.getParamsToStr("sourceMsg").replaceAll("\n", "").replaceAll("\r", "").replaceAll(";", "；"), "数据同步已下线", "ERROR");
                }
            }
        }
    }
}